package com.seaboxdata.jt809.kafka;

import com.alibaba.fastjson.JSONObject;
import com.seaboxdata.jt809.common.util.PropertiesUtil;
import com.seaboxdata.jt809.db.JT809Dao;
import com.seaboxdata.jt809.packet.JT809Packet0x1202;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

/**
 * @ClassName MessageConsumer
 * @Description 消息消费者类用于将消息取出kafka
 * @Author DFJX
 * @Company 北京东方金信科技有限公司
 * @Date 2020/7/20 17:34
 *         Version 1.0
 **/
public class JT809Packet0x1202MessageConsumer {

    private static final boolean isWin = System.getProperty("os.name").toLowerCase().contains("win");

    private String kafkaServer = PropertiesUtil.getProperty("kafka.server");

    private String authConfig = PropertiesUtil.getProperty("kafka.jaas.config");

    private String authWinConfig = PropertiesUtil.getProperty("kafka.win.jaas.config");

    private String krb5Config = PropertiesUtil.getProperty("krb5.config");

    private String krb5WinConfig = PropertiesUtil.getProperty("krb5.win.config");

    private String isKerberes = PropertiesUtil.getProperty("kafka.kerberos.enable");

    private String topicGroup = PropertiesUtil.getProperty("kafka.topic.group");

    private String topic = PropertiesUtil.getProperty("kafka.topic");

    public void getMessage() {

        Properties props = new Properties();
        if ("true".equals(isKerberes)) {
            if (isWin) {
                //如果是windows指定本地的配置文件,本地调试使用,将配置文件指定在本地
                this.authConfig = authWinConfig;
                this.krb5Config = krb5WinConfig;
            }
            System.setProperty("java.security.auth.login.config", authConfig);
            System.setProperty("java.security.krb5.conf", krb5Config);
            props.put("security.protocol", "SASL_PLAINTEXT");
            props.put("sasl.kerberos.service.name", "kafka");
            props.put("sasl.mechanism", "GSSAPI");
        }
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, topicGroup);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Collections.singletonList(topic));
        List<JT809Packet0x1202> list = new ArrayList<JT809Packet0x1202>();
        long begin = System.currentTimeMillis();
        long batchSize = Long.valueOf(PropertiesUtil.getProperty("batch.size")) == 0 ? 10000
                : Long.valueOf(PropertiesUtil.getProperty("batch.size"));
        long interval = Long.valueOf(PropertiesUtil.getProperty("interval")) == 0 ? 10
                : Long.valueOf(PropertiesUtil.getProperty("interval"));
        while (true) {
            ConsumerRecords<String, String> messages = consumer.poll(100L);
            for (ConsumerRecord<String, String> message : messages) {
                list.add(JSONObject.parseObject(message.value(), JT809Packet0x1202.class));
                long end = System.currentTimeMillis();
                if ((end - begin) / 1000 >= interval) {
                    JT809Dao.batchAdd0x1202(list);
                    list = new ArrayList<JT809Packet0x1202>();
                    begin = System.currentTimeMillis();
                }
            }
            long end = System.currentTimeMillis();
            if ((end - begin) / 1000 >= interval && list.size() > 0) {
                JT809Dao.batchAdd0x1202(list);
                list = new ArrayList<JT809Packet0x1202>();
                begin = System.currentTimeMillis();
            }
        }
    }
}
